#include "DatagramClient.h"
#include "DatagramSocketChannel.h"
#include "EventLoop.h"
#include "IpV4Address.h"
#include "IpV6Address.h"
#include "Manager.h"
#include "PlatLogger.h"
#include "SocketOpt.h"
#include "UdpSocket.h"
#include "Util.h"
#include "XnccAddress.h"
#include "XnccClient.h"
#include "XnccType.h"
namespace xncc {
namespace foundation {

    DatagramClient::DatagramClient(EventLoop* loop, std::shared_ptr<XnccAddress> serverAddr)
        : loop_(loop),
          connect_(true),
          nextConnId_(1),
          serverAddr_(std::move(serverAddr)),
          serverAddrStr_(serverAddr_->toStr())
    {
    }

    /// 长期驻留内存的设计
    DatagramClient::~DatagramClient() = default;

    void DatagramClient::init()
    {
        loop_->registerUserFunc(shared::InnerMsgId::ON_PUB_IPC_MSG,
                                [this](const char* msg, uint32_t len, types::thread_id_t tid,
                                       types::connection_id_t connId) { handleIoMsg(msg, len, tid, connId); });
    }

    void DatagramClient::connect()
    {
        int sockfd = -1;
        int ret    = 0;
        if (serverAddr_->Type() == shared::XNCC_ADDR_TYPE::IPV4_UDP) {
            const auto v4Addr = std::dynamic_pointer_cast<IpV4Address>(serverAddr_);
            sockfd            = SocketOpt::createNonblockingOrDie(serverAddr_->family(), IPPROTO_UDP, SOCK_DGRAM);
            ret = ::connect(sockfd, v4Addr->getSockAddr(), static_cast<socklen_t>(sizeof(struct sockaddr_in)));
        }
        else if (serverAddr_->Type() == shared::XNCC_ADDR_TYPE::IPV6_UDP) {
            const auto v6Addr = std::dynamic_pointer_cast<IpV6Address>(serverAddr_);
            sockfd            = SocketOpt::createNonblockingOrDie(serverAddr_->family(), IPPROTO_UDP, SOCK_DGRAM);
            ret = ::connect(sockfd, v6Addr->getSockAddr(), static_cast<socklen_t>(sizeof(struct sockaddr_in6)));
        }
        (void)ret;
        connect_ = true;
        TRACE_INFO << "engine id:"_s << loop_->id() << " client start, connect server:"_s << serverAddrStr_;
        loop_->runInLoop([this, sockfd]() { newConnection(sockfd); });
    }

    void DatagramClient::disconnect() { connect_ = false; }

    void DatagramClient::stop() { connect_ = false; }
    void DatagramClient::handleIoMsg(const char* msg,
                                     uint32_t    len,
                                     types::thread_id_t /*tid*/,
                                     types::connection_id_t connId)
    {
        if (len > MAX_UDP_MSG_LEN) {
            throw shared::logic_exception(" msg too bigger");
            return;
        }
        if (unlikely(serverChannel_ == nullptr)) {
            TRACE_ERROR << "connid disconnected:"_s << connId;
            return;
        }
        if (connIdFd_[connId] != serverChannel_->fd()) {
            TRACE_ERROR << "msg sender set error connid:"_s << connId;
            return;
        }
        serverChannel_->sendMsg(msg, len);
    }

    void DatagramClient::newConnection(int sockfd)
    {
        const auto connId = shared::CommonBusi::generateConnId();
#ifdef XNCC_DEBUG
        loop_->assertInLoopThread(__FUNCTION__);
#endif
        TRACE_INFO << " engine id:"_s << loop_->id() << " connect server success, connid:"_s << connId
                   << " server addr:"_s << serverAddrStr_;
        QUEUE_MGR.pubIPCDataToBusiThread(serverAddrStr_.c_str(), static_cast<uint32_t>(serverAddrStr_.size()),
                                         shared::InnerMsgId::ON_CONNECTED, loop_->id(), connId);
        serverChannel_ = std::make_shared<DatagramSocketChannel>(loop_, sockfd, connId);
        serverChannel_->init();
        serverChannel_->setCloseCallback(
            [this, connId](const std::string& reason) { removeConnection(connId, reason); });
        serverChannel_->enableReading();
        connIdFd_[connId] = sockfd;
    }

    void DatagramClient::removeConnection(types::connection_id_t connId, const std::string& disconnectedReason)
    {
        if (connIdFd_.find(connId) == connIdFd_.end()) {
            TRACE_ERROR << " engine id:"_s << loop_->id() << " has call disconnected, connid:"_s << connId
                        << " server addr:"_s << serverAddrStr_;
            return;
        }
        TRACE_INFO << " engine id:"_s << loop_->id() << " disconnected, connid:"_s << connId << " server addr:"_s
                   << serverAddrStr_;
        QUEUE_MGR.pubIPCDataToBusiThread(disconnectedReason.c_str(), static_cast<uint32_t>(disconnectedReason.size()),
                                         shared::InnerMsgId::ON_DISCONNECTED, loop_->id(), connId);
        serverChannel_->disableAll();
        serverChannel_->remove();
        serverChannel_.reset();
        connIdFd_.erase(connId);
        if (retry_) {
            TRACE_INFO << "retry-Retry connecting to"_s << serverAddr_->toStr() << " after "_s << retryDelayMs_
                       << " milliseconds"_s;
            std::this_thread::sleep_for(std::chrono::milliseconds(retryDelayMs_));
            connect();
        }
    }
}  // namespace foundation
}  // namespace xncc